package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * @author Felix
 * @date 2022/12/17
 * 查询维度的工具类
 */
public class DimUtil {

    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName,Tuple2.of("id",id));
    }
    /**
     * 查询维度数据
     * 优化：旁路缓存
     * 旁路缓存思路：先从缓存中查询维度，如果在缓存中找到了要关联的维度，直接将缓存中的维度作为返回值进行返回(缓存命中)；
     *             如果在缓存中没有找到要关联的维度数据，发送请求到Phoenix表中去查询维度，并将查询的维度放到缓存中，方便下次查询使用。
     * 缓存产品选择：
     *      状态：     性能好     维护性差
     *      Redis：    性能也不差     维护性好     √
     * Redis相关设置：
     *      Key：    dim:维护表名:主键1_主键2
     *      数据类型：String
     *      expire: 1day
     *      注意：如果Phoenix中维度数据发生了变化，需要将redis中缓存的数据清掉
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String> ... columnNameAndValues) {
        //拼接从Redis中查询维度数据的key
        StringBuilder redisKey = new StringBuilder("dim:"+tableName.toLowerCase()+":");

        //拼接查询语句
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + " where ");

        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            redisKey.append(columnValue);
            selectSql.append(columnName + "='" + columnValue + "'");
            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        Jedis jedis = null;
        String dimInfoStr = null;
        JSONObject dimInfoJsonObj = null;
        try {
            jedis = RedisUtil.getJedis();
            dimInfoStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }

        if(StringUtils.isNotEmpty(dimInfoStr)){
            //缓存命中  直接将缓存中的维度作为返回值进行返回
            dimInfoJsonObj = JSON.parseObject(dimInfoStr);
        }else{
            //在缓存中没有找到对应的维度   发送请求到Phoenix表中去查询维度，并将查询的维度放到缓存中，方便下次查询使用
            System.out.println("从Phoenix表中查询维度数据的SQL:" + selectSql);
            //底层还是调用PhoenixUtil中的查询方法
            List<JSONObject> dimInfoJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
            if (dimInfoJsonObjList != null && dimInfoJsonObjList.size() > 0) {
                //如果是根据主键去查询维度，维度数据只能有一条
                dimInfoJsonObj = dimInfoJsonObjList.get(0);
                // 将查询的维度放到缓存中，方便下次查询使用
                if(jedis != null){
                    jedis.setex(redisKey.toString(),24*3600,dimInfoJsonObj.toJSONString());
                }
            } else {
                System.out.println("在Phoenix表中没有找到对应的维度数据");
            }
        }

        //关闭连接
        if(jedis != null){
            System.out.println("~~关闭Jedis客户端~~");
            jedis.close();
        }

        return dimInfoJsonObj;
    }

    public static void deleteCached(String tableName,String id){
        String redisKey = "dim:"+tableName.toLowerCase()+":" + id;

        Jedis jedis = null;
        try {
            jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("从Redis中删除维度数据失败~~~");
        } finally {
            if(jedis != null){
                System.out.println("~~从Redis中删除缓存数据后，关闭Jedis客户端~~");
                jedis.close();
            }
        }
    }

    //从phoenix表中查询维度
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String> ... columnNameAndValues) {
        //拼接查询语句
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + " where ");

        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "'");
            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
            }
        }

        System.out.println("从Phoenix表中查询维度数据的SQL:" + selectSql);

        //底层还是调用PhoenixUtil中的查询方法
        List<JSONObject> dimInfoJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        JSONObject dimInfoJsonObj = null;
        if (dimInfoJsonObjList != null && dimInfoJsonObjList.size() > 0) {
            //如果是根据主键去查询维度，维度数据只能有一条
            dimInfoJsonObj = dimInfoJsonObjList.get(0);
        } else {
            System.out.println("在Phoenix表中没有找到对应的维度数据");
        }
        return dimInfoJsonObj;
    }

    public static void main(String[] args) {
        // System.out.println(getDimInfoNoCache("DIM_BASE_TRADEMARK", Tuple2.of("id","1")));
        // System.out.println(getDimInfo("DIM_BASE_TRADEMARK", Tuple2.of("id","1")));
        System.out.println(getDimInfo("DIM_BASE_TRADEMARK", "1"));
    }
}
